package com.xs.consumer;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @Title: <P>Consumer.java</P>
 * @Package <P>com.xs.consumer</P>
 * @Description: <P>TODO该类...</P>
 * @author renweiping
 * @date 2019年3月11日 下午1:01:20
 * @version 1.0
 */

public class Consumer extends Thread  {
	private static final Logger logger = LoggerFactory.getLogger(Consumer.class);
	private static final String topic ="test2";
	private static final String brokerUrl = "127.0.0.1:9092";
	private static KafkaConsumer<String,String> consumer=null;
	private String  groupId;
	
	
	/**
	 * 
	 */
	public Consumer(String groupId) {
		this.groupId = groupId;
		Properties config = initConfig();
		consumer = new KafkaConsumer<String,String>(config);
		consumer.subscribe(Arrays.asList(topic));
	}
	
	
	/**
	 * @Description: TODO
	 * @param @return   
	 * @return Properties  
	 * @throws
	 * @author renweiping
	  *@e_mail renweiping_1113@google.com
	 * @date 2019年3月11日
	 */
	
	private  Properties initConfig() {
		Properties config = new Properties();
		config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
		config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
		config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
		config.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
		config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
		config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
		return config;
	}
	

	
	
	
	@Override
	public void run(){
		while(true){
			ConsumerRecords<String, String> records =  consumer.poll(100);
			for(ConsumerRecord<String, String> record :   records){
				logger.info("KafkaConsumer.poll(\"{}\", {}, {}, {}, \"{}\") OK.",
                        record.topic(), record.partition(), record.offset(),
                        record.key(), record.value());
				consumer.commitAsync();
			}
		}
		
		
	}
	


}
